package com.macaline.technology.rabbitMq;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.xnx3.DateUtil;

/**
 * MQ消息延迟队列工具类
 * 
 * @author warm_joy
 */
public class DelayMQUtil {
	
	private static String messageId = null;
	private static Channel channel = null;
	private static Connection connection = null;
	private static String delayExchangeName = "delay_MESSAGEID_exchange";
	private static String delayQueueName = "delay_MESSAGEID_queue";
	private static String delayQueueRoutKey = "delay_MESSAGEID_rout_key";
	private static String delayComsuerExchangeName = "delay_comsuery_MESSAGEID_exchange";
	private static String delayComsuerQueueName = "delay_comsuer_MESSAGEID_queue";

	/**
	 * 发送延迟消息
	 * 
	 * @author warm_joy
	 * @param messageId 消息ID
	 * @param message   发送消息内容
	 * @param delayTime 延迟的时间，单位：秒(s)
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void sendMessage(String messageId, String message, int delayTime)
			throws IOException, TimeoutException {
		// 初始换信息
		DelayMQUtil.messageId = messageId;
		initializeQueue();
		/*
		 *  设置发送消息的过期时间和持久化状态
		 *  deliveryMode：1：不持久化；2：持久化
		 *  expiration：消息过期时间，单位：毫秒（ms）
		 */
		AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
		builder.expiration((delayTime * 1000) + "").deliveryMode(2);
		// 发送消息
		getChannel().basicPublish(delayExchangeName, delayQueueRoutKey, builder.build(), message.getBytes());
		getChannel().close();
		getConnection().close();
	}
	
	/**
	 * 初始化交换机、队列、路由等信息
	 * 
	 * @author warm_joy
	 * @throws IOException
	 * @throws TimeoutException
	 */
	private static void initializeQueue() throws IOException, TimeoutException {
		// 初始化交换机、队列、路由名称
		delayExchangeName = "delay_" + messageId + "_exchange";
		delayQueueName = "delay_" + messageId + "_queue";
		delayComsuerExchangeName = "delay_comsuery_" + messageId + "_exchange";
		delayComsuerQueueName = "delay_comsuer_" + messageId + "_queue";
		delayQueueRoutKey = "delay_" + messageId + "_rout_key";
		// 消费者交换机，队列绑定
		getChannel().exchangeDeclare(delayComsuerExchangeName, BuiltinExchangeType.DIRECT);
		getChannel().queueDeclare(delayComsuerQueueName, true, false, false, null);
		getChannel().queueBind(delayComsuerQueueName, delayComsuerExchangeName, delayQueueRoutKey);
		// 延迟交换机，队列绑定
		getChannel().exchangeDeclare(delayExchangeName, BuiltinExchangeType.DIRECT);
		// 为队列添加消息过期后使用的私信交换机名称
		Map<String, Object> args = new HashMap<String, Object>();
		args.put("x-dead-letter-exchange", delayComsuerExchangeName);
		args.put("x-dead-letter-routing-key", delayQueueRoutKey);
		getChannel().queueDeclare(delayQueueName, true, false, false, args);
		getChannel().queueBind(delayQueueName, delayExchangeName, delayQueueRoutKey);
	}

	/**
	 * 接收MQ消息并进行消息处理
	 * 
	 * @author warm_joy
	 * @param messageId 消息ID
	 * @param action    进行消息处理的逻辑接口 接口参数“msg”为接收到的MQ消息，可在此接口内方法对消息进行逻辑处理
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void recrviceMsg(String messageId, MqConsumerProcessor action) throws IOException, TimeoutException {
		// 获取接收消息的消息队列
		String queueName = "delay_comsuer_" + messageId + "_queue";
		Connection connection = RabbitMqUtil.getConnection();
		Channel channel = connection.createChannel();
		channel.queueDeclare(queueName, true, false, false, null);
		DefaultConsumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				String msg = new String(body);
				action.messageProcessor(msg);
			}
		};
		channel.basicConsume(queueName, true, consumer);
	}

	/**
	 * 获取一个channel
	 * @author warm_joy
	 * @return
	 * @throws IOException
	 * @throws TimeoutException
	 */
	private static Channel getChannel() throws IOException, TimeoutException {
		if(channel == null || channel.isOpen() == false) {
			channel = getConnection().createChannel();
		}
		return channel;
	}

	/**
	 * 获取一个连接
	 * @author warm_joy
	 * @return
	 * @throws IOException
	 * @throws TimeoutException
	 */
	private static Connection getConnection() throws IOException, TimeoutException {
		if (connection == null || connection.isOpen() == false) {
			connection = RabbitMqUtil.getConnection();
		}
		return connection;
	}

	public static void main(String[] args) throws IOException, TimeoutException {
		DelayMQUtil.sendMessage("test", "我发送了延迟消息" + DateUtil.currentDate(DateUtil.FORMAT_DEFAULT), 5);
		DelayMQUtil.sendMessage("test", "我发送了延迟消息" + DateUtil.currentDate(DateUtil.FORMAT_DEFAULT), 10);
		recrviceMsg("test", new MqConsumerProcessor() {
			@Override
			public void messageProcessor(String msg) {
				System.out
						.println("delay_comsuer_test_queue接收消息：" + msg + "----"
								+ DateUtil.currentDate(DateUtil.FORMAT_DEFAULT));
			}
		});
	}

}
